package com.xzy.rabbit.demo.confirmpublish;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 发布确认
 *
 * @author xzy
 * @date 2021/9/2914:59
 */
@Slf4j
@AllArgsConstructor
class Producer implements Runnable {
    private final Connection connection;
    private final String exchangeName;
    private final String routingKey;
    private final Integer messageCount;
    private final long workTime;

    public void produce() throws IOException, TimeoutException, InterruptedException {
        // 1.获取信道
        try (Channel channel = connection.createChannel()) {
            String currentThread = Thread.currentThread().getName();

            // 2.开启发布确认
            channel.confirmSelect();

            // 3.设置回调方法
            final ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
                log.info(currentThread + " ACK  deliveryTag：{} multiple：{}\n", deliveryTag, multiple);
            };
            final ConfirmCallback nachCallback = (deliveryTag, multiple) -> {
                log.info(currentThread + " NACK  deliveryTag：{} multiple：{}\n", deliveryTag, multiple);
            };
            channel.addConfirmListener(ackCallback, nachCallback);

            // 4.发送消息
            for (int i = 0; i < messageCount; i++) {
                Thread.sleep(workTime);
                String message = "msg" + i;
                channel.basicPublish(exchangeName, routingKey, null, message.getBytes());
                log.info(currentThread + " 成功发送：`{}`----{}---->[{}]\n", message, routingKey, exchangeName);
            }
        }
    }

    @Override
    public void run() {
        try {
            produce();
        } catch (IOException | TimeoutException | InterruptedException e) {
            log.error(e.getMessage(), e);
            Thread.currentThread().interrupt();
        }
    }
}
